package com.chenjl.comsumer;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.chenjl.Constants;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;
import kafka.serializer.StringDecoder;
import kafka.utils.VerifiableProperties;
/**
 * 消费者
 * Old Consumer Client API 消费者直连zk，然后消费
 * 
 * 2016-9-12 15:34:53
 * @author chenjinlong
 */
public class Consumer01 {
	private static final Logger logger = LoggerFactory.getLogger(Consumer01.class);
	
	public static void main(String[] args) {
		Properties props = new Properties();
		
		props.put("zookeeper.connect",Constants.ZK_URL);
		props.put("zookeeper.session.timeout.ms","5000");
		props.put("zookeeper.sync.time.ms","200");
		
		props.put("auto.commit.interval.ms","1000");
		props.put("auto.offset.reset","smallest");
		
		props.put("serializer.class","kafka.serializer.StringEncoder");
		
		//消费组
		props.put("group.id","test");
		
		ConsumerConfig config = new ConsumerConfig(props);
		ConsumerConnector consumer = Consumer.createJavaConsumerConnector(config);
		
		Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(Constants.TOPIC_NAME,new Integer(1));

        StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
        StringDecoder valueDecoder = new StringDecoder(new VerifiableProperties());

        Map<String,List<KafkaStream<String, String>>> consumerMap = consumer.createMessageStreams(topicCountMap,keyDecoder,valueDecoder);
        KafkaStream<String,String> stream = consumerMap.get(Constants.TOPIC_NAME).get(0);
        
        ConsumerIterator<String,String> iterator = stream.iterator();
        while (iterator.hasNext()) {
        	MessageAndMetadata<String, String> message = iterator.next();
        	logger.info("key  :  {}  , message : {}",message.key(),message.message());
        }
	}
}